package cn.com.netedge.iot.frontend.datasave.utils;

import javax.annotation.PostConstruct;

import cn.com.netedge.iot.frontend.datasave.config.DataSaveConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import cn.com.netedge.iot.common.constants.CacheKeyConstants;
import cn.com.netedge.iot.common.constants.Globals;
import cn.com.netedge.iot.common.constants.MqConstants;
import cn.com.netedge.iot.common.utils.ChronicleQueueUtil;
import cn.com.netedge.iot.common.utils.RedisUtil;
import cn.com.netedge.iot.common.utils.StringUtil;

@Component
public class MqConsumeUtil {
    private static final Logger logger = LoggerFactory.getLogger(MqConsumeUtil.class);

    @PostConstruct
    public void init() {
    }

    private static void insertCacheOrQueue(String cacheKey, String queueName, String body) {
        long len = RedisUtil.llen(cacheKey);
        if (len > Globals.MAX_CACHE_SIZE) {//超出缓存队列的最大存放数，则重新放入消息队列中，避免缓存溢出
            logger.warn("{} exceed max cache size:{}, current size: {}", cacheKey, len);

            ChronicleQueueUtil.enqueue(queueName, body);
            return;
        }
        //存入缓存处理分表
        RedisUtil.rightPush(cacheKey, body);
    }

    public static void consumeDataCOnlineMessage(String queueName, String msg, String cacheKeyName) {
        insertCacheOrQueue(cacheKeyName + DataSaveConfig.getMachineName(), queueName, msg);
    }

    public static void consumeCEventMessage(String queueName, String msg, String cacheKeyName) {
        insertCacheOrQueue(cacheKeyName + DataSaveConfig.getMachineName(), queueName, msg);
    }

    public static void consumeFreezeDataMessage(String queueName, String msg, String cacheKeyName) {
        insertCacheOrQueue(cacheKeyName + DataSaveConfig.getMachineName(), queueName, msg);
    }

    @Async("asyncServiceExecutor")
    public static void consume(String dest, String msg) {
        if (StringUtil.isEmpty(dest)) {
            return;
        }
        long start = System.currentTimeMillis();
        if (MqConstants.QUEUE_DATA_C_ONLINE_MESSAGE.equals(dest)) {
            MqConsumeUtil.consumeDataCOnlineMessage(dest, msg, CacheKeyConstants.KEY_DATA_C_ONLINE);
        } else if (MqConstants.QUEUE_DATA_C_LOAD_MESSAGE.equals(dest)) {
            MqConsumeUtil.consumeFreezeDataMessage(dest, msg, CacheKeyConstants.KEY_DATA_C_LOAD);
        } else if (MqConstants.QUEUE_DATA_C_LOAD2_MESSAGE.equals(dest)) {
            MqConsumeUtil.consumeFreezeDataMessage(dest, msg, CacheKeyConstants.KEY_DATA_C_LOAD2);
        } else if (MqConstants.QUEUE_DATA_C_DAILY_MESSAGE.equals(dest)) {
            MqConsumeUtil.consumeFreezeDataMessage(dest, msg, CacheKeyConstants.KEY_DATA_C_DAILY);
        } else if (MqConstants.QUEUE_DATA_C_MONTHLY_MESSAGE.equals(dest)) {
            MqConsumeUtil.consumeFreezeDataMessage(dest, msg, CacheKeyConstants.KEY_DATA_C_MONTHLY);
        } else {
            logger.error("dlq -> queue_name:{} ignore...", dest);
        }
        long end = System.currentTimeMillis();
        logger.info("收到 {} 消息:{}, 耗时:{} 秒", dest, msg.toString(), (end - start) / 1000);
    }

}
